package org.atlantis.datasink.server;

import java.net.SocketAddress;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import org.atlantis.datasink.DeviceData;
import org.atlantis.datasink.event.EventManager;
import org.atlantis.datasink.event.EventQueue;
import org.atlantis.datasink.server.buildin.ActiveConnectManager;
import org.atlantis.datasink.utils.InternalEventNames;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.handler.timeout.IdleStateAwareChannelUpstreamHandler;
import org.jboss.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@SuppressWarnings("unchecked")
public class DataIoHandler extends IdleStateAwareChannelUpstreamHandler {

	private static final Logger log = LoggerFactory.getLogger(DataIoHandler.class);
	private boolean isServer = false;
	private boolean isClient = false;
	private SocketAddress remoteAddr;
	private EventQueue disconnectQueue = EventManager.getInstance().getOrCreate(InternalEventNames.DEVICE_DISCONNECT);
	private EventQueue rawDataQueue = EventManager.getInstance().getOrCreate(InternalEventNames.RAW_DATA_RECEIVED);
	private Set<Long> deviceIds = new HashSet<Long>();

	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
		List<DeviceData> datas = (List<DeviceData>) e.getMessage();
		if (datas == null || datas.isEmpty()) return;
		for (DeviceData data : datas) {
			long deviceId = data.getDeviceId();
			if (deviceIds.contains(deviceId)) continue;
			deviceIds.add(deviceId);
		}
		rawDataQueue.offer(datas);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
		String side = getSide();
		log.error("Remote {}: {} error. Caused by: {}", new Object[] { side, remoteAddr, e.getCause() });
	}

	@Override
	public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {

		Class<?> channelType = ctx.getChannel().getClass();

		isServer = channelType.equals(Class.forName("org.jboss.netty.channel.socket.nio.NioAcceptedSocketChannel"));
		isClient = channelType.equals(Class.forName("org.jboss.netty.channel.socket.nio.NioClientSocketChannel"));

		remoteAddr = ctx.getChannel().getRemoteAddress();
		if (isClient) {
			log.info("Active model server connected to remote server: {}", remoteAddr);
		} else if (isServer) {
			log.info("Remote client: {} connected to passive model server.", remoteAddr);
		} else {
			log.info("Remote point: {} connected server", remoteAddr);
		}
	}

	@Override
	public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
		disconnectQueue.offer(new HashSet<Long>(deviceIds));
		deviceIds.clear();
		if (isClient) {
			log.warn("Disconnect from server: {}", remoteAddr);
			ActiveConnectManager.getInstance().waitToReconnect(remoteAddr);
		} else if (isServer) {
			log.warn("Remote client: {} disconnected.", remoteAddr);
		} else {
			log.warn("Remote point: {} disconnected.", remoteAddr);
		}
	}

	@Override
	public void channelIdle(ChannelHandlerContext ctx, IdleStateEvent e) throws Exception {
		log.warn("Remote {}: {} accidental death. So close it.", getSide(), remoteAddr);
		ctx.getChannel().close();
	}

	private String getSide() {
		String side = "point";
		if (isClient) side = "server";
		else if (isServer) side = "client";
		return side;
	}
}
